package com.xmim.server.examples;

import com.xmim.core.constants.MessageCode;
import com.xmim.core.container.ServerContainer;
import com.xmim.core.entity.Message;
import com.xmim.core.exception.ContainerException;
import com.xmim.fulcrum.FulcrumFactory;
import com.xmim.fulcrum.node.FulcrumNode;
import com.xmim.fulcrum.node.FulcrumServiceNode;
import com.xmim.fulcrum.session.AbstractFulcrumSession;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.TimeUnit;

/**
 * create by CAOYOU
 **/
@Slf4j
public class NettyFulcrumSession extends AbstractFulcrumSession<String, ChannelHandler> {


    private Channel channel;
    private Bootstrap bootstrap;
    private EventLoopGroup group;


    public NettyFulcrumSession(FulcrumFactory factory, ServerContainer parent, FulcrumServiceNode node) {
        super(factory, parent, node);
    }


    @Override
    protected void onInit() {
        this.bootstrap = new Bootstrap();
        this.group = new NioEventLoopGroup();
        this.bootstrap.group(group);
        this.bootstrap.channel(NioSocketChannel.class);
        this.bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        this.bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        // 设置通道初始化
        this.bootstrap.handler(new NettyChannelInitializer(this.getFactory(), this.handlers));
    }

    @Override
    protected void onBind() throws ContainerException {
        if (connect) {
            return;
        }
        // 服务器ip地址
        String host = node.getHost();
        // 服务器端口
        int port = node.getPort();
        try {
            log.debug("分布式客户端开始连接:{}", getNode().toString());
            ChannelFuture connect = bootstrap.connect(host, port);
            connect.addListener(channelFuture);
            this.channel = connect.channel();
        } catch (Exception e) {
            throw new ContainerException("分布式客户端连接失败!", e);
        }
    }


    ChannelFutureListener channelFuture = new ChannelFutureListener() {

        @Override
        public void operationComplete(ChannelFuture channelFuture) throws Exception {

            if (channelFuture.isSuccess()) {
                log.debug("分布式客户端:连接成功");
                connect = true;
                channel = channelFuture.channel();
                //发送链接成功的通知
                Message<FulcrumNode> message = Message.<FulcrumNode>
                        builder()
                        .messageType(MessageCode.MSG_FULCRUM_HEARTBEAT)
                        .body(getNode())
                        .build();
                writeAndFlush(message);
            } else {
                log.debug("分布式客户端:每隔2s重连....");
                channelFuture.channel().eventLoop().schedule(new Runnable() {

                    @Override
                    public void run() {
                        try {
                            onBind();
                        } catch (ContainerException e) {
                            log.debug("分布式客户端:重新连接失败", e);
                        }
                    }
                }, 2, TimeUnit.SECONDS);
            }
        }
    };

    @Override
    protected void onClose() throws ContainerException {
        try {
            group.shutdownGracefully();
        } catch (Exception e) {
            throw new ContainerException("分布式客户端关闭失败!", e);
        }
    }


    @Override
    public boolean checkHandler(Class<?> handler) {
        return ChannelHandler.class.isAssignableFrom(handler.getClass());
    }

    @Override
    public NettyFulcrumSession writeAndFlush(Object pkg) {
        if (!connect) {
            log.error("分布式客户端未连接:" + getNode().toString());
            return this;
        }
        channel.writeAndFlush(pkg);
        return this;
    }


}
